package com.miao.im.biz;

import cn.hutool.core.collection.CollectionUtil;
import com.miao.im.common.Constant;
import com.miao.im.common.Message;
import com.miao.im.common.annotation.MessageCommand;
import com.miao.im.common.enums.Command;
import com.miao.im.common.enums.EventType;
import com.miao.im.common.enums.TargetType;
import com.miao.im.dto.SocketMessageDto;
import com.miao.im.model.AuthResponseModel;
import com.miao.im.model.P2PRequestModel;
import com.miao.im.model.P2PResponseModel;
import com.miao.im.repository.IUserRepository;
import com.miao.im.server.session.SessionContext;
import com.miao.im.server.session.SessionManager;
import com.miao.im.service.message.ChatMessage;
import com.miao.im.support.disruptor.DisruptorEvent;
import com.miao.im.support.handler.AbstractMessageHandler;
import com.miao.im.support.mq.IProducerMessageService;
import com.miao.im.support.pulisher.IEventProducer;
import com.miao.im.support.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/3/1 15:12
 * @Description:
 */
@Component
@MessageCommand(Command.P2P_SENDER)
@Slf4j
public class P2PRequestHandler extends AbstractMessageHandler<P2PRequestModel.P2PRequest> {
    @Autowired
    private IUserRepository userRepository;
    @Autowired
    private IProducerMessageService messageService;
    @Autowired
    private IEventProducer disruptorProducer;


    @Override
    protected void doRead(P2PRequestModel.P2PRequest body, Message message) {
        Message request = Message.writeRequest(message.getVersion(), message.getSequence(), Command.P2P_NOTIFY.getCode(), body.toByteArray());
        SessionManager.sendToUser(String.valueOf(body.getToUserId()), request);
    }

    @Override
    protected Mono<Void> doDispatch(P2PRequestModel.P2PRequest body, Message message, SessionContext sessionContext) {
        log.info("接收到单聊请求:fromUserId={},toUserId={}", body.getFromUserId(), body.getToUserId());
        Long toUserId = body.getToUserId();
        return userRepository.reactiveGetOnlineUsers(toUserId).doOnSuccess(onlineUsers -> {
            if (CollectionUtil.isEmpty(onlineUsers)) {
                //没有在线用户的时候回执一个假的ack消息
                this.publishOfflineMsg(body, message);
                Message R2SAck = Message.writeRequest(message.getVersion(), message.getSequence(), Command.P2P_NOTIFY_TO_SENDER_ACK.getCode(), body.toByteArray());
                SessionManager.send(sessionContext, R2SAck);
            } else {
                for (AuthResponseModel.AuthResponse onlineUser : onlineUsers) {
                    //本机不走mq
                    this.sendToUser(body, message, onlineUser);
                }
            }
            //回执一个确认消息
            P2PResponseModel.P2PResponse p2PResponse = P2PResponseModel.P2PResponse.newBuilder()
                    .setFromUserId(body.getFromUserId())
                    .setFromRoutingKey(body.getFromRoutingKey())
                    .setSendTimestamp(System.currentTimeMillis())
                    .build();
            Message response = Message.writeResponse(message.getVersion(), message.getSequence(), Command.P2P_SENDER.getCode(), p2PResponse.toByteArray());
            SessionManager.send(sessionContext, response);
        }).then();
    }

    private void sendToUser(P2PRequestModel.P2PRequest body, Message message, AuthResponseModel.AuthResponse onlineUser) {
        if (Constant.DISPATCHER_ROUTING_KEY_NAME.equals(onlineUser.getRoutingKey())) {
            this.doRead(body, message);
        } else {
            SocketMessageDto messageDto = new SocketMessageDto(message);
            messageService.send(Constant.DISPATCHER_EXCHANGE, onlineUser.getRoutingKey(), JsonUtil.toJSONStr(messageDto));
        }
    }


    private void publishOfflineMsg(P2PRequestModel.P2PRequest body, Message message) {
        ChatMessage ch = ChatMessage.builder()
                .fromUserId(body.getFromUserId())
                .fromUserNickname(body.getFromUserNickname())
                .fromUserAvatar(body.getFromUserAvatar())
                .targetId(body.getToUserId())
                .targetName(body.getToUserNickname())
                .targetAvatar(body.getToUserAvatar())
                .targetType(TargetType.PERSON.getCode())
                .sequence(message.getSequence())
                .mediaType(body.getMediaType())
                .content(body.getMessage())
                .expand(body.getExpand())
                .sendTime(LocalDateTime.now())
                .build();
        disruptorProducer.publish(new DisruptorEvent<>(EventType.SAVE_OFFLINE_MSG, ch));
    }


}
